package cn.edu.flink.tutorial.window;

import cn.edu.flink.tutorial.source.UDSourceFunction;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


public class TimeWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Tuple2<Long, Long>> streamSource = env.addSource(new UDSourceFunction()).map(x -> Tuple2.of(x, 1L)).returns(Types.TUPLE(Types.LONG, Types.LONG));

        SingleOutputStreamOperator<Tuple2<Long, Long>> WatermarkedStream = streamSource
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Tuple2<Long, Long>>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<Long, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple2<Long, Long> element, long recordTimestamp) {
                                return element.f0;
                            }
                        }));

        AllWindowedStream<Tuple2<Long, Long>, org.apache.flink.streaming.api.windowing.windows.TimeWindow> windowedStream = WatermarkedStream
                // .timeWindowAll(Time.seconds(10));
                // .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));
                // .timeWindowAll(Time.seconds(10), Time.seconds(5));
                // .windowAll(EventTimeSessionWindows.withGap(Time.seconds(5)));
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        windowedStream
                .reduce(new ReduceFunction<Tuple2<Long, Long>>() {
                            @Override
                            public Tuple2<Long, Long> reduce(Tuple2<Long, Long> value1, Tuple2<Long, Long> value2) throws Exception {
                                return Tuple2.of(value1.f0, value2.f0);
                            }
                        },
                        new ProcessAllWindowFunction<Tuple2<Long, Long>, Object, org.apache.flink.streaming.api.windowing.windows.TimeWindow>() {
                            @Override
                            public void process(Context context, Iterable<Tuple2<Long, Long>> elements, Collector<Object> out) throws Exception {
                                System.out.println("currentTimeMillis:" + System.currentTimeMillis());
                                System.out.println("window:" + context.window());
                                System.out.println("elements:" + elements);
                                System.out.println();
                            }
                        });

        env.execute();
    }

}
